// Copyright 2013 Google Inc. All Rights Reserved.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package com.google.api.ads.dfa.matchtables;

import com.google.api.ads.dfa.axis.factory.DfaServices;
import com.google.api.ads.dfa.axis.v1_20.AdBase;
import com.google.api.ads.dfa.axis.v1_20.AdRecordSet;
import com.google.api.ads.dfa.axis.v1_20.AdRemote;
import com.google.api.ads.dfa.axis.v1_20.AdSearchCriteria;
import com.google.api.ads.dfa.axis.v1_20.AdType;
import com.google.api.ads.dfa.axis.v1_20.ApiException;
import com.google.api.ads.dfa.axis.v1_20.ClickThroughUrl;
import com.google.api.ads.dfa.axis.v1_20.ClickTracker;
import com.google.api.ads.dfa.axis.v1_20.CreativeAd;
import com.google.api.ads.dfa.axis.v1_20.CreativeAssignment;
import com.google.api.ads.dfa.axis.v1_20.CreativeGroup;
import com.google.api.ads.dfa.axis.v1_20.CreativeGroupRecordSet;
import com.google.api.ads.dfa.axis.v1_20.CreativeGroupRemote;
import com.google.api.ads.dfa.axis.v1_20.CreativeGroupSearchCriteria;
import com.google.api.ads.dfa.axis.v1_20.DefaultAd;
import com.google.api.ads.dfa.axis.v1_20.MobileAd;
import com.google.api.ads.dfa.axis.v1_20.PlacementAssignment;
import com.google.api.ads.dfa.axis.v1_20.RotationGroup;
import com.google.api.ads.dfa.axis.v1_20.TargetableAdBase;
import com.google.api.ads.dfa.axis.v1_20.TrackingAd;
import com.google.api.ads.dfa.lib.client.DfaSession;

import com.google.api.ads.dfa.matchtables.CreativeTypes.CreativeRotationType;

import java.rmi.RemoteException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;

/**
 * Implementation of LookupObject that will generate Ad match tables. Designed to be
 * component of AdCreativeLookup to be able to exchange data with CreativeSubLookup.
 *
 * @author Moshe Kahan
 */
class AdSubLookup implements LookupObject {

  // Main aggregation maps
  // Ads per campaign
  private Map<Long, Set<Long>> campaignToAdIdsMap;
  private Map<Long, Long> campaignToAdvertisersMap;

  // Secondary lookup maps
  private Map<Long, String> adTypeMap;
  private Map<Long, Long> creativeToRenderingMap;

  // Size helper
  SizeLookupHelper sizeLookupHelper;

  // DT columns that this class is interested in
  private int adColumnIndex;
  private int advertiserColumnIndex;
  private int campaignColumnIndex;
  private int orderIdColumnIndex;

  // MatchTable settings and SOAP remote objects
  private TableSettings tableSettings;
  private final ThreadLocal<AdRemote> adService;
  private final ThreadLocal<CreativeGroupRemote> creativeGroupService;

  // Multi-Threading support.
  private ExecutorService executor;

  /**
   * Constructor.  Besides the DFA service/session objects needed to create SOAP remote objects to
   * the API, constructor also accepts matchTableTypes to determine if both Ad and Creative match
   * tables should be generated.
   *
   * @param dfaServices DFA services object needed to create SOAP remotes to the API
   * @param session DFA session object needed to create SOAP remotes to the API
   * @param tableSettings match table application configuration used as a MatchTable object factory
   */
  AdSubLookup(DfaServices dfaServices, DfaSession session,
      TableSettings tableSettings, ExecutorService executor, SizeLookupHelper sizeLookupHelper,
      Map<Long, Long> creativeToRenderingMap) {
    // Request the services from the service client factory.
    this(AdCreativeLookup.initializeService(dfaServices, session, AdRemote.class),
        AdCreativeLookup.initializeService(dfaServices, session, CreativeGroupRemote.class),
        tableSettings, executor, sizeLookupHelper, creativeToRenderingMap);
  }

  AdSubLookup(ThreadLocal<AdRemote> adService,
      ThreadLocal<CreativeGroupRemote> creativeGroupService, TableSettings tableSettings,
      ExecutorService executor, SizeLookupHelper sizeLookupHelper,
      Map<Long, Long> creativeToRenderingMap) {
    // Request the services from the service client factory.
    this.adService = adService;
    this.creativeGroupService = creativeGroupService;
    this.tableSettings = tableSettings;

    campaignToAdIdsMap = new HashMap<Long, Set<Long>>();
    campaignToAdvertisersMap = new HashMap<Long, Long>();
    this.creativeToRenderingMap = creativeToRenderingMap;
    adTypeMap = new HashMap<Long, String>();

    // Set up Executor to run concurrent DFA API retrievals.
    this.executor = executor;

    // Size lookup helper
    this.sizeLookupHelper = sizeLookupHelper;

    adColumnIndex = -1;
    advertiserColumnIndex = -1;
    campaignColumnIndex = -1;
    orderIdColumnIndex = -1;
  }

  /**
   * Called when LookupObjectManager is parsing the DT file header. Method stores index
   * of ad, advertiser, campaign, and creative columns.
   *
   * @param index in DT file columns
   * @param header name of this DT file field/column
   */
  public void addHeader(int index, String header) {
    if (LookupField.AdId.getFieldName().equals(header)) {
      adColumnIndex = index;
    } else if (LookupField.AdvertiserId.getFieldName().equals(header)) {
      advertiserColumnIndex = index;
    } else if (LookupField.BuyId.getFieldName().equals(header)) {
      campaignColumnIndex = index;
    } else if (LookupField.OrderId.getFieldName().equals(header)) {
      orderIdColumnIndex = index;
    }
  }

  /**
   * Called to clear the DT header information' usually because you're processing a new file.
   */
  public void clearHeaders() {
    adColumnIndex = -1;
    advertiserColumnIndex = -1;
    campaignColumnIndex = -1;
    orderIdColumnIndex = -1;
  }

  /**
   * Called after DT file line has been tokenized into array of Strings.  Aggregate over 4 column
   * of interest.  Only aggregate creative id if Creative match table is to be generated. Create 
   * grouping of ads IDs per campaign to make it easier for later SOAP API lookups.
   *
   * @param values array of tokenized Strings of DT file line
   */
  public void addValues(String[] values) {
    // Can only parse if Ad-ID, Advertiser-ID, and Buy-ID are present.
    if (advertiserColumnIndex > -1) {
      String advertiserId = values[advertiserColumnIndex];
      if (campaignColumnIndex > -1 && AdCreativeLookup.isValid(advertiserId)) {
        long numericAdvertiserId = Long.valueOf(advertiserId);
        String campaignId = values[campaignColumnIndex];
        if (!AdCreativeLookup.isValid(campaignId) && orderIdColumnIndex > -1) {
          campaignId = values[orderIdColumnIndex];
        }
        if (AdCreativeLookup.isValid(campaignId)) {
          long numericCampaignId = Long.valueOf(campaignId);
          if (adColumnIndex > -1) {
            String adId = values[adColumnIndex];
            if (AdCreativeLookup.isValid(adId)) {
              campaignToAdvertisersMap.put(numericCampaignId, numericAdvertiserId);
              AdCreativeLookup.addToSetToMap(campaignToAdIdsMap, numericCampaignId,
                  Long.valueOf(adId));
            }
          }
        }
      }
    }
  }

  /**
   * Called when parsing/aggregation is complete. Will lookup via SOAP all unique DFA entries
   * in the DT file and pass data to MatchTable class.
   *
   * @param outputPath location of match table to be generated.
   */
  public void generateMatchTable(String outputPath)
      throws MatchTableFileException, DfaRemoteException {
    List<MatchTable> matchTableList = new ArrayList<MatchTable>();
    MatchTable adMatchTable =
        AdCreativeLookup.createMatchTable(tableSettings, MatchTable.Type.AD, matchTableList);
    MatchTable adPageAssignmentsMatchTable =
        AdCreativeLookup.createMatchTable(tableSettings, MatchTable.Type.AD_PAGE_ASSIGNMENTS,
            matchTableList);
    MatchTable creativeAssignmentMatchTable =
        AdCreativeLookup.createMatchTable(tableSettings, MatchTable.Type.CREATIVE_AD_ASSIGNMENTS,
            matchTableList);
    MatchTable keyvalMatchTable =
        AdCreativeLookup.createMatchTable(tableSettings, MatchTable.Type.KEYVAL, matchTableList);
    for (MatchTable matchTable : matchTableList) {
      matchTable.open(outputPath);
    }

    Progress.printProgress(adMatchTable, adPageAssignmentsMatchTable,
        creativeAssignmentMatchTable, keyvalMatchTable);

    final List<AdBase> adResults = new ArrayList<AdBase>();
    // Do the raw ad entity retrieval using concurrent threads.
    List<Future<?>> taskResults = new ArrayList<Future<?>>();
    final Progress progress = new Progress();
    progress.start(campaignToAdIdsMap.keySet().size());
    System.out.println("Ad Retrieval Phase");
    for (Map.Entry<Long, Set<Long>> entry : campaignToAdIdsMap.entrySet()) {
      final Long campaignId = entry.getKey();
      final Set<Long> adSet = entry.getValue();
      taskResults.add(executor.submit(new Callable<Void>() {
        public Void call() throws DfaRemoteException {
          retrieveAds(campaignId, adSet, adResults);
          progress.next();
          return null;
        }
      }));
    }
    AdCreativeLookup.waitForFutures(taskResults);
    progress.end();

    System.out.println("Ad Secondary Retrieval Phase");
    // API retrievals
    sizeLookupHelper.retrieveAdSizes(adResults);
    retrieveAdTypes();

    // Generate ad match table.  Store IDs of various fields in the Ad entity.
    // Creative Groups
    Map<Long, Set<Long>>  advertiserCreativeGroupSetMap = new HashMap<Long, Set<Long>>();
    for (Long advertiserId : campaignToAdvertisersMap.values()) {
      advertiserCreativeGroupSetMap.put(advertiserId, new HashSet<Long>());
    }
    // Creative Assignments
    Map<AdBase, Set<CreativeAssignment>> adCreativeAssignmentSetMap =
        new HashMap<AdBase, Set<CreativeAssignment>>();
    generateAdMatchTables(adResults, adCreativeAssignmentSetMap, advertiserCreativeGroupSetMap,
        adMatchTable, adPageAssignmentsMatchTable, keyvalMatchTable);

    // Retrieve creative group entities concurrently.
    final Map<Long, String> creativeGroupMap = new HashMap<Long, String>();
    taskResults.clear();
    progress.start(advertiserCreativeGroupSetMap.keySet().size());
    for (Map.Entry<Long, Set<Long>> entry : advertiserCreativeGroupSetMap.entrySet()) {
      final long advertiserId = entry.getKey();
      final Set<Long> creativeGroupIdSet = entry.getValue();
      taskResults.add(executor.submit(new Callable<Void>() {
        public Void call() throws DfaRemoteException {
          retrieveCreativeGroups(advertiserId, creativeGroupIdSet, creativeGroupMap);
          progress.next();
          return null;
        }
      }));
    }
    AdCreativeLookup.waitForFutures(taskResults);
    progress.end();

    if (creativeAssignmentMatchTable != null) {
      generateCreativeAssignmentMatchTable(adCreativeAssignmentSetMap, creativeGroupMap,
          creativeAssignmentMatchTable);
    }

    for (MatchTable matchTable : matchTableList) {
      matchTable.close();
    }
  }

  // Generates all ad lookups and match table rows for all ads in one campaign.
  private void generateAdMatchTables(List<AdBase> adList,
      Map<AdBase, Set<CreativeAssignment>> adCreativeAssignmentSetMap,
      Map<Long, Set<Long>> advertiserCreativeGroupSetMap, MatchTable adMatchTable,
      MatchTable adPageAssignmentsMatchTable, MatchTable keyvalMatchTable)
          throws MatchTableFileException {
    for (AdBase result : adList) {
      Long advertiserId = campaignToAdvertisersMap.get(result.getCampaignId());
      String adId = Long.toString(result.getId());
      Set<CreativeAssignment> creativeAssignmentSet = new HashSet<CreativeAssignment>();
      Set<Long> creativeGroupSet = advertiserCreativeGroupSetMap.get(advertiserId);
      extractCreatives(result, creativeAssignmentSet, creativeGroupSet);
      adCreativeAssignmentSetMap.put(result, creativeAssignmentSet);
      if (adMatchTable != null) {
        adMatchTable.write(
            Long.toString(advertiserId),
            Long.toString(result.getCampaignId()),
            adId,
            result.getName(),
            extractClickThruURL(result),
            adTypeMap.get(result.getTypeId()),
            sizeLookupHelper.getSize(result.getSizeId()),
            result.getComments());
      }
      PlacementAssignment[] placementAssignments = result.getPlacementAssignments();
      if (adPageAssignmentsMatchTable != null && placementAssignments != null) {
        for (PlacementAssignment placementAssignment : placementAssignments) {
          adPageAssignmentsMatchTable.write(
              adId,
              Long.toString(placementAssignment.getPlacementId()));
        }
      }
      if (keyvalMatchTable != null) {
        String expression = extractKeywordExpression(result);
        if (expression != null) {
          keyvalMatchTable.write(
              adId,
              expression);
        }
      }
    }

    if (adMatchTable != null) {
      adMatchTable.flush();
    }
    if (keyvalMatchTable != null) {
      keyvalMatchTable.flush();
    }
    if (adPageAssignmentsMatchTable != null) {
      adPageAssignmentsMatchTable.flush();
    }
  }

  // Generate creative assignment match table for all ads in this advertiser/campaign.
  private void generateCreativeAssignmentMatchTable(
      Map<AdBase, Set<CreativeAssignment>> creativeAssignmentMap,
      Map<Long, String> creativeGroupMap, MatchTable creativeAssignmentMatchTable)
          throws MatchTableFileException {
    for (Map.Entry<AdBase, Set<CreativeAssignment>> entry : creativeAssignmentMap.entrySet()) {
      AdBase ad = entry.getKey();
      String rotationType = "";
      if (ad instanceof RotationGroup) {
        long rotationTypeId = ((RotationGroup) ad).getRotationType();
        for (CreativeRotationType creativeRotationType : CreativeRotationType.values()) {
          if (creativeRotationType.getId() == rotationTypeId) {
            rotationType = creativeRotationType.getName();
            break;
          }
        }
      }
      SimpleDateFormat dateFormat = new SimpleDateFormat("MM-dd-yyyy HH:mm:ss");
      for (CreativeAssignment creativeAssignment : entry.getValue()) {
        // Only output creative-ad-assignments whose rendering ID is in the DT file
        if (creativeToRenderingMap.containsKey(creativeAssignment.getCreativeId())) {
          long renderingId = creativeToRenderingMap.get(creativeAssignment.getCreativeId());
          String startDate = "";
          String endDate = "";
          if (creativeAssignment.getStartDate() != null) {
            startDate = dateFormat.format(creativeAssignment.getStartDate().getTime());
          } else if (ad.getStartTime() != null) {
            startDate = dateFormat.format(ad.getStartTime().getTime());
          }
          if (creativeAssignment.getEndDate() != null) {
            endDate = dateFormat.format(creativeAssignment.getEndDate().getTime());
          } else if (ad.getEndTime() != null) {
            endDate = dateFormat.format(ad.getEndTime().getTime());
          }
          creativeAssignmentMatchTable.write(
              Long.toString(ad.getId()),
              Long.toString(renderingId),
              startDate,
              endDate,
              rotationType,
              creativeGroupMap.get(
                  creativeAssignment.getCreativeGroupAssignment().getCreativeGroup1Id()),
              creativeGroupMap.get(
                  creativeAssignment.getCreativeGroupAssignment().getCreativeGroup2Id()),
              creativeAssignment.getClickThroughUrl() == null ? "" :
                creativeAssignment.getClickThroughUrl().getComputedClickThroughUrl());
        }
      }
    }
    creativeAssignmentMatchTable.flush();
  }

  // Extract methods to retrieve field information often by using instanceOf

  // Utility method to extract click-thru URL property of Ad.
  private String extractClickThruURL(AdBase ad) {
    if (ad instanceof ClickTracker) {
      ClickTracker clickTracker = (ClickTracker) ad;
      ClickThroughUrl url = clickTracker.getClickThroughUrl();
      if (url != null) {
        return url.getComputedClickThroughUrl();
      }
    }
    return null;
  }

  // Utility method to retrieve creative IDs from Ad by trying to downcast object to Ad
  // subclass that supports creative or creativeAssignment property.
  private void extractCreatives(AdBase ad,
      Set<CreativeAssignment> creativeAssignmentSet,
      Set<Long> creativeGroupSet) {
    CreativeAssignment[] creativeAssignmentArray = null;
    if (ad instanceof DefaultAd) {
      creativeAssignmentArray = new CreativeAssignment[1];
      creativeAssignmentArray[0] = ((DefaultAd) ad).getCreativeAssignment();
    } else if (ad instanceof CreativeAd) {
      CreativeAd creativeAd = (CreativeAd) ad;
      creativeAssignmentArray = new CreativeAssignment[1];
      CreativeAssignment dummyCreativeAssignment = new CreativeAssignment();
      dummyCreativeAssignment.setCreativeId(creativeAd.getCreativeId());
      dummyCreativeAssignment.setCreativeGroupAssignment(creativeAd.getCreativeGroupAssignment());
      dummyCreativeAssignment.setClickThroughUrl(creativeAd.getClickThroughUrl());
      dummyCreativeAssignment.setEndDate(creativeAd.getEndTime());
      dummyCreativeAssignment.setStartDate(creativeAd.getStartTime());
      creativeAssignmentArray[0] = dummyCreativeAssignment;
    } else if (ad instanceof MobileAd) {
      creativeAssignmentArray = ((MobileAd) ad).getCreativeAssignments();
    } else if (ad instanceof RotationGroup) {
      creativeAssignmentArray = ((RotationGroup) ad).getCreativeAssignments();
    } else if (ad instanceof TrackingAd) {
      creativeAssignmentArray = ((TrackingAd) ad).getCreativeAssignments();
    }
    if (creativeAssignmentArray != null) {
      for (CreativeAssignment creativeAssignment : creativeAssignmentArray) {
        extractCreativeGroups(creativeAssignment, creativeGroupSet);
        creativeAssignmentSet.add(creativeAssignment);
      }
    }
  }

  private void extractCreativeGroups(CreativeAssignment creativeAssignment,
      Set<Long> creativeGroups) {
    long creativeGroupId =
        creativeAssignment.getCreativeGroupAssignment().getCreativeGroup1Id();
    if (creativeGroupId > 0) {
      creativeGroups.add(creativeGroupId);
    }
    creativeGroupId =
        creativeAssignment.getCreativeGroupAssignment().getCreativeGroup2Id();
    if (creativeGroupId > 0) {
      creativeGroups.add(creativeGroupId);
    }
  }

  // Extract keyword expression from Ad if downcasted type supports that property.
  private String extractKeywordExpression(AdBase ad) {
    if (ad instanceof TargetableAdBase) {
      return ((TargetableAdBase) ad).getKeywordExpression();
    }
    if (ad instanceof MobileAd) {
      return ((MobileAd) ad).getKeywordExpression();
    }
    return null;
  }

  // Retrieval methods
  // Thread-Safe
  private void retrieveAds(Long campaignId, Set<Long> adIds, List<AdBase> adResults)
      throws DfaRemoteException {
    List<AdBase> adList = new ArrayList<AdBase>();
    SoapPageableLookup<AdBase, AdSearchCriteria> adSoapLookup =
        new SoapPageableLookup<AdBase, AdSearchCriteria>(adIds,
            new AdSoapLookupCallback(campaignId));
    while (adSoapLookup.hasNext()) {
      AdBase result = adSoapLookup.next();
      adList.add(result);
    }
    addAds(adResults, adList);
  }

  private void retrieveAdTypes() throws DfaRemoteException {
    SoapLookup<AdType> adTypeSoapLookup = new SoapLookup<AdType>(new AdTypeSoapLookupCallback());
    while (adTypeSoapLookup.hasNext()) {
      AdType result = adTypeSoapLookup.next();
      adTypeMap.put(result.getId(), result.getName());
    }
  }

  // Thread-Safe
  private void retrieveCreativeGroups(Long advertiserId, Set<Long> creativeGroupIdSet,
      Map<Long, String> creativeGroupMap) throws DfaRemoteException {
    List<CreativeGroup> creativeGroups = new ArrayList<CreativeGroup>();
    SoapLookup<CreativeGroup> creativeGroupSoapLookup =
        new SoapLookup<CreativeGroup>(creativeGroupIdSet,
            new CreativeGroupSoapLookupCallback(advertiserId));
    while (creativeGroupSoapLookup.hasNext()) {
      creativeGroups.add(creativeGroupSoapLookup.next());
    }
    addCreativeGroups(creativeGroups, creativeGroupMap);
  }

  // Adder methods to ensure thread safety for specific retrieve methods

  private void addAds(List<AdBase> adResults, List<AdBase> ads) {
    synchronized (adResults) {
      adResults.addAll(ads);
    }
  }

  private void addCreativeGroups(List<CreativeGroup> creativeGroups,
      Map<Long, String> creativeGroupMap) {
    synchronized (creativeGroupMap) {
      for (CreativeGroup creativeGroup : creativeGroups) {
        creativeGroupMap.put(creativeGroup.getId(), creativeGroup.getName());
      }
    }
  }

  // SoapLookupCallbacks

  private class AdTypeSoapLookupCallback implements SoapLookupCallback<AdType> {
    public int getRecords(long[] ids, List<AdType> recordList)
        throws ApiException, RemoteException {
      // Retrieve the possible ad types.
      AdType[] adTypes = adService.get().getAdTypes();
      // Display ad type and its ID.
      for (AdType result : adTypes) {
        adTypeMap.put(result.getId(), result.getName());
      }
      return adTypes.length;
    }
  }

  private class AdSoapLookupCallback
    implements SoapPageableLookupCallback<AdBase, AdSearchCriteria> {

    AdSoapLookupCallback(long campaignId) {
      this.campaignId = campaignId;
    }

    public int getRecords(AdSearchCriteria criteria, List<AdBase> recordList)
        throws ApiException, RemoteException {
      AdRecordSet adRecordSet = adService.get().getAds(criteria);
      for (AdBase ad : adRecordSet.getRecords()) {
        recordList.add(ad);
      }
      return adRecordSet.getTotalNumberOfPages();
    }

    public AdSearchCriteria getCriteria(long[] ids) {
      AdSearchCriteria adSearchCriteria = new AdSearchCriteria();
      adSearchCriteria.setCampaignId(campaignId);
      if (ids != null) {
        adSearchCriteria.setIds(ids);
      }
      return adSearchCriteria;
    }

    private long campaignId;
  }

  private class CreativeGroupSoapLookupCallback implements SoapLookupCallback<CreativeGroup> {

    private long[] advertiserIds;

    CreativeGroupSoapLookupCallback(Long advertiserId) {
      this.advertiserIds = new long[1];
      advertiserIds[0] = advertiserId;
    }

    public int getRecords(long[] ids, List<CreativeGroup> recordList)
        throws ApiException, RemoteException {

      CreativeGroupSearchCriteria searchCriteria = new CreativeGroupSearchCriteria();
      searchCriteria.setIds(ids);
      searchCriteria.setAdvertiserIds(advertiserIds);

      CreativeGroupRecordSet recordSet =
          creativeGroupService.get().getCreativeGroups(searchCriteria);
      CreativeGroup[] creativeGroupArray = recordSet.getRecords();
      for (CreativeGroup entity : creativeGroupArray) {
        recordList.add(entity);
      }
      return creativeGroupArray.length;
    }
  }
}
